package com.transport.analysis.traffic;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.Calendar;

public class TrafficMapper
        extends Mapper<LongWritable, TrafficRecordWritable, Text, Text> {

    private final Text compositeKey = new Text();
    private final Text passengersAndCrowd = new Text();

    @Override
    protected void map(LongWritable key, TrafficRecordWritable value, Context context)
            throws IOException, InterruptedException {

        // No specific filtering for bus/subway here, as the model aims for both
        int routeId = value.getRouteId();
        String stopId = value.getStopId();
        Timestamp timestamp = value.getTimestamp();
        int passengersOn = value.getPassengersOn();

        if (stopId == null || stopId.isEmpty()) {
            return; // Skip records with no stop ID
        }
        if (timestamp == null) {
            return; // Skip records with no timestamp
        }

        // Extract hour from timestamp
        Calendar cal = Calendar.getInstance();
        cal.setTimeInMillis(timestamp.getTime());
        int hour = cal.get(Calendar.HOUR_OF_DAY);

        // Composite Key: (线路ID, 站点ID, 小时)
        compositeKey.set(routeId + "_" + stopId + "_" + hour);

        // Value: (上车人数, 拥挤度) - using passengersOn as a proxy for crowd level
        passengersAndCrowd.set(passengersOn + "\t" + passengersOn);

        try {
            context.write(compositeKey, passengersAndCrowd);
        } catch (Exception e) {
            System.err.println(
                    "An unexpected error occurred for record: " + value.toString() + ": " + e.getMessage());
        }
    }
}